package com.opennews.openplatform;

import com.opennews.openplatform.dto.OrderDto;
import com.opennews.openplatform.flink.GenericDeserializationSchema;
import com.opennews.openplatform.util.NacosConfigLoader;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Map;

public class Main {
    public static void main(String[] args) {
        try {
            Map<String, Object> config = NacosConfigLoader.loadKafkaConfig();

            // Sets up the streaming environment.
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

            // Set the restart strategy to fixed delay, e.g., 3 retries with 10 seconds delay
            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000));

            // Creates a Kafka source.
            KafkaSource<OrderDto> kafkaSource = KafkaSource.<OrderDto>builder()
                    .setTopics("order-topic")
                    .setBootstrapServers((String) config.get("bootstrap-servers"))
                    //                    .setBootstrapServers("127.0.0.1:9092")
                    .setGroupId("flink-job")
                    .setStartingOffsets(OffsetsInitializer.latest())
                    .setValueOnlyDeserializer(new GenericDeserializationSchema<>(OrderDto.class))
                    .build();

            // Creates a stream that reads from Kafka using the KafkaSource.
            DataStream<OrderDto> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "KafkaSource");

            // Process the stream to call an external service when a specific message is received
            stream.map(value -> {
                System.out.println("Order id: " + value.getId());

                return value;
            });

            env.execute("FlinkKafkaJob");

            System.out.println("Done");
        } catch (Exception ex) {
            System.out.println(ex.getMessage());
        }
    }
}